// -*-mode:c++; coding:utf-8-*-
#ifndef _XBASE_BLOCK_CACHE_UNIT_HPP_
#define _XBASE_BLOCK_CACHE_UNIT_HPP_

// unused
#if 0

#include <ext/hash_map>
#include <boost/pool/detail/singleton.hpp>
#include <boost/thread/mutex.hpp>

#include <logger.hpp>

#include "table_schema_manager.hpp"
#include "table_schema.hpp"

#include "basic_block_cache_manager.hpp"

namespace xbase {

	class block_cache_unit
	{
	public:
		typedef basic_block_cache_manager<block_cache_unit> block_cache_manager_1;
		typedef boost::details::pool::singleton_default<block_cache_manager_1> block_cache_manager;
		
		// typedef unit_key;
		// typedef unit_key_hasher;

		// typedef unit_node;
		// typedef unit_block_key;
		// typedef unit_block_key;
	public:
		typedef __gnu_cxx::hash_map<cache_key,
					    cache_node*,
					    cache_key_hasher,
					    cache_key_equal_to> node_map_type;

		bool commit(cache_node *node);
		bool commit_and_checkout(cache_node *node);
		cache_node *checkout(const cache_key &key);
		// inc ref count, 并调整结点位置
		bool checkout(cache_node *node);
		bool checkin(cache_node *node);

		enum { DEFAULT_SLOT_NUMBER = 10000 };

		block_cache_unit(const unit_key &key)
			: m_key(key)
			, m_last_used_time(cur_seconds())
			, m_list()
			, m_water_line(NULL)
			, m_block_data_size(0)
			, m_block_number(0)
			, m_node_map(DEFAULT_SLOT_NUMBER) {}

		~block_cache_unit() {
			boost::mutex::scoped_lock guard(m_lock);
			clear_cache_until_empty();
		}

		uint32_t cached_block_number() const {
			return m_block_number;
		}
		uint32_t cached_block_size() const {
			return m_block_data_size;
		}

		struct is_in_sstable {
			const uint64_t m_fileid;

			is_in_sstable(const uint64_t &fileid)
				: m_fileid(fileid) {}
			
			bool operator()(cache_node *node) const {
				return m_fileid == node->m_key.m_file_id;
			}
		};
		// 遍历所有未使用块，释放属于指定SSTable的块。因为要求
		// 用户保证释放时没有任何块被使用，所以这样做是可以的。
		void clear_sstable_cache(const uint64_t &file_id) {
			boost::mutex::scoped_lock guard(m_lock);
			clear_cache_if_until_not(is_in_sstable(file_id),
						 unused());

			// 注意！这里有一个隐患！
			// 
			// 当清除sstable的块缓存时，只能清除掉没被使用
			// 的块缓存，因此sstable析构前必须要确保所有的
			// 块缓存都没有被使用才行！否则，会导致块缓存
			// checkin的时候出错，而无法释放，造成内存泄漏！
			//
			// 因此在这里做一下检查，如果有上面的错误现象出
			// 现，程序core掉。
#ifndef NDEBUG
			for(cache_node_list_type::iterator iter = m_list.begin();
			    iter != m_list.end(); ++iter)
			{
				if(iter->m_key.m_file_id == file_id)
				{
					debug_dump_cache_list("clear_sstable_cache", &(*iter));
					LOG(FATAL, "block_cache_unit::clear_sstable_cache "
					    << file_id << "'s cache#"
					    << iter->m_key.m_block_index
					    << " still in use, ref_count = "
					    << iter->m_value.ref_count());
					assert(false);
				}
			}
#endif	// NDEBUG
		}

		void clear_unused_cache() {
			LOG_ENV(int count_block = m_block_number);
			LOG_ENV(int count_bytes = m_block_data_size);

			boost::mutex::scoped_lock guard(m_lock);
			clear_cache_until_busy_block();

			LOG_ENV(count_block = count_block - m_block_number);
			LOG_ENV(count_bytes = count_bytes - m_block_data_size);
			LOG(INFO, "block_cache_unit::clear_unused_cache clear "
			    << count_block << " block cache(s) with "
			    << count_bytes << " bytes, and left "
			    << m_block_number << " block cache(s) with "
			    << m_block_data_size << " bytes");
		}

		const unit_key &key() const {
			return m_key;
		}

		void check_eliminate_for_timeout();

	protected:
		cache_node *get_node(const cache_key &key) const {
			cache_node *node = 0;
			const node_map_type::const_iterator iter = m_node_map.find(key);
			if(iter != m_node_map.end())
			{
				node = iter->second;
			}
			return node;
		}

		void push_front(cache_node *node) {
			m_list.push_front(*node);
		}

		void push_back(cache_node *node) {
			if(m_water_line == NULL)
			{
				m_list.push_back(*node);
			}
			else
			{
				cache_node_list_type::iterator water_line = m_list.iterator_to(*m_water_line);
				m_list.insert(water_line, *node);
			}
			m_water_line = node;
		}

		void move_to_front(cache_node *node) {
			assert(node->is_linked());
			if(node == m_water_line)
			{
				cache_node_list_type::iterator cur = m_list.iterator_to(*node);
				++cur;
				if(cur == m_list.end())
				{
					m_water_line = NULL;
				}
				else
				{
					m_water_line = m_list.get_real_value_traits().to_value_ptr(cur.pointed_node());
				}
			}
			if(&m_list.front() != node)
			{
				node->unlink();
				m_list.push_front(*node);
			}
		}

		void move_to_back(cache_node *node) {
			assert(node->is_linked());
			cache_node_list_type::iterator cur = m_list.iterator_to(*node);
			++cur;
			if((cur != m_list.end())
			   && (m_water_line != m_list.get_real_value_traits().to_value_ptr(cur.pointed_node())))
			{
				node->unlink();
				if(m_water_line == NULL)
				{
					m_list.push_back(*node);
				}
				else
				{
					cache_node_list_type::iterator water_line = m_list.iterator_to(*m_water_line);
					m_list.insert(water_line, *node);
				}
				m_water_line = node;
			}
			else
			{
				m_water_line = node;
			}
		}

		void check_eliminate();

		table_schema::group_info::pointer get_group_info() const;
		table_schema::pointer get_schema() const;
		
		// 在列表中，从后向前，遍历所有块，对每个满足条件
		// PredIf的块，执行删除、释放操作，直到某个块不满足
		// PredNotQuit条件时退出
		//
		// 注意：该实现与 list 的定义和实现很有关系，必须要很小
		// 心的维护！
		//
		// TODO: 以后的改进过程中，尽量替换当前的实现方式，现在
		// 尽管效率高，但依赖性太强，不安全。
		template<typename PredIf, typename PredNotQuit>
		inline
		void clear_cache_if_until_not(const PredIf &pred_if,
					      const PredNotQuit &pred_not_quit) {
			const cache_node_list_type::node_ptr root_node = m_list.end().pointed_node(); // 这是双向链表的根结点
			cache_node_list_type::node_ptr pos = cache_node_list_type::node_traits::get_previous(root_node); // get tail
			while(pos != root_node)
			{
				cache_node * const node = m_list.get_real_value_traits().to_value_ptr(pos); // 获取到cache_node的指针
				// 获取到当前的指针后，就指向下一个元素，必须要删除该结点之前就指向下一个要处理的元素
				pos = cache_node_list_type::node_traits::get_previous(pos);

				if(! pred_not_quit(node))
				{
					break;
				}

				if(! pred_if(node)) // go next
				{
					continue;
				}

				// 更新water_line
				if(node == m_water_line)
				{
					cache_node_list_type::node_ptr cur = cache_node_list_type::node_traits::get_next(pos);
					cache_node_list_type::node_ptr new_water_line = cache_node_list_type::node_traits::get_next(cur);
					if(new_water_line == root_node)
					{
						m_water_line = NULL;
					}
					else
					{
						m_water_line = m_list.get_real_value_traits().to_value_ptr(new_water_line);
					}
				}

				// 必须先从列表中删除，再释放内存！
				node->unlink();

				// update counter
				m_block_number --;
				m_block_data_size -= node->m_value.m_block_size;
				assert((m_block_number >= 0)
				       && (m_block_data_size >= 0));

				const node_map_type::size_type num = m_node_map.erase(node->m_key);
				assert(num == 1); (void)num;
				// LOG_IF((num != 1), FATAL,
				//        "block_cache_unit::clear_cache_until_not erase cache_key{"
				//        << node->m_key.m_file_id << ","
				//        << node->m_key.m_block_index
				//        << "} failed(" << num
				//        << "), please CHECK it!");
				
				// 重要！
				// 删除cache_key的时候，一定要记录通知
				// 缓存管理器，来删除那里面关于这条key
				// 的可能存在的索引项。
				block_cache_manager::instance().deallocate(node);
			}
		}

		struct always_true {
			bool operator()(cache_node *node) const {
				return true;
			}
		};
		// 在列表中，从后向前，依次删除块缓存，直到不满足指定条
		// 件时退出
		template<typename Pred>
		inline
		void clear_cache_until_not(const Pred &pred) {
			clear_cache_if_until_not(always_true(), pred);
		}

		// 析构时调用，所有块缓存的引用数一定为0，删除所有缓存
		struct false_with_assert {
			bool operator()(cache_node *node) const {
				LOG(ERROR, "block_cache_unit::false_with_assert node still in use: "
				    << node->m_key.m_file_id << "#"
				    << node->m_key.m_block_index << ";");
				assert(node->m_value.is_no_ref());
				return false;
			}
		};
		void clear_cache_until_empty() {
			clear_cache_until_not(false_with_assert());
		}

		// 清除所有未被使用的缓存，只剩下正在使用的块
		struct unused {
			bool operator()(cache_node *node) const {
				return node->m_value.is_no_ref();
			}
		};
		void clear_cache_until_busy_block() {
			clear_cache_until_not(unused());
		}

		// 清除未被使用的缓存，直到块数据的字节数少于指定值
		struct unused_and_has_more_bytes {
			volatile std::size_t *m_block_data_size;
			const std::size_t m_bytes;
			
			unused_and_has_more_bytes(volatile std::size_t *block_data_size,
						  std::size_t bytes)
				: m_block_data_size(block_data_size)
				, m_bytes(bytes) {}
			
			bool operator()(cache_node *node) const {
				return ((*m_block_data_size > m_bytes)
					&& node->m_value.is_no_ref());
			}
		};
		void clear_cache_until_busy_block_or_data_bytes_not_great_than(std::size_t bytes) {
			clear_cache_until_not(unused_and_has_more_bytes(&m_block_data_size, bytes));
		}

		void update_time() {
			m_last_used_time = cur_seconds();
		}

		bool is_timeout();

#ifndef NDEBUG_BLOCK_CACHE
		void debug_dump_cache_list(const char *name, cache_node *node);
#else
		void debug_dump_cache_list(const char *, cache_node *) {}
#endif	// NDEBUG_BLOCK_CACHE

	private:
		unit_key m_key;
		// 上次使用该缓存模块的时间，若超过设置时间时间还没有被使用，则淘汰时就可以将所有没被引用的块淘汰
		::time_t m_last_used_time;
		boost::mutex m_lock;

		cache_node_list_type m_list;
		// 指向第一个引用计数为0的结点，后续的结点引用计数变为
		// 0后插入到该结点之前，并更新该指针；淘汰时还是从后向
		// 前。这可以使得淘汰的结点是最老不使用的，刚刚不使用的
		// 结点会优先保留下来。
		//
		// 初始为NULL
		cache_node *m_water_line;

		std::size_t m_block_data_size;
		std::size_t m_block_number;
		// cached_data<table_schema::group_info> m_group_info;
	
		node_map_type m_node_map;

		mutable table_schema::weak_pointer m_schema;
		mutable boost::mutex m_schema_lock;

		mutable table_schema::group_info::weak_pointer m_group_info;
		mutable boost::mutex m_group_info_lock;
	};

} // namespace xbase

#endif	// 0

#endif	// _XBASE_BLOCK_CACHE_UNIT_HPP_
